package com.lhd.app.dwd;

import com.lhd.common.utils.MyKafkaUtil;
import com.lhd.common.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradePayDetailSuc2 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(905));

        tableEnv.executeSql(MyKafkaUtil.getTopicDb("pay_detail_suc_211126"));

        Table paymentInfo = tableEnv.sqlQuery("select " +
                "data['user_id'] user_id, " +
                "data['order_id'] order_id, " +
                "data['payment_type'] payment_type, " +
                "data['callback_time'] callback_time, " +
                "`pt` " +
                "from topic_db " +
                "where `table` = 'payment_info' " +
                "and `type` = 'update' " +
                "and data['payment_status']='1602'");
        tableEnv.createTemporaryView("payment_info", paymentInfo);

        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail( " +
                "id string, " +
                "order_id string, " +
                "user_id string, " +
                "sku_id string, " +
                "sku_name string, " +
                "sku_num string, " +
                "order_price string, " +
                "province_id string, " +
                "activity_id string, " +
                "activity_rule_id string, " +
                "coupon_id string, " +
                "create_time string, " +
                "source_id string, " +
                "source_type_id string, " +
                "source_type_name string, " +
                "split_activity_amount string, " +
                "split_coupon_amount string, " +
                "split_total_amount string, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail","pay_detail_suc_order_211126"));

        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        // 首先创建一个包含正确时间属性的视图
        Table resultTable = tableEnv.sqlQuery("" +
                "select " +
                "od.id order_detail_id, " +
                "od.order_id, " +
                "od.user_id, " +
                "od.sku_id, " +
                "od.sku_name, " +
                "od.province_id, " +
                "od.activity_id, " +
                "od.activity_rule_id, " +
                "od.coupon_id, " +
                "pi.payment_type payment_type_code, " +
                "dic.dic_name payment_type_name, " +
                "pi.callback_time, " +
                "od.source_id, " +
                "od.source_type_id, " +
                "od.source_type_name, " +
                "od.sku_num, " +
                "od.order_price, " +
                "od.split_activity_amount, " +
                "od.split_coupon_amount, " +
                "od.split_total_amount split_payment_amount, " +
                "od.row_op_ts row_op_ts " +
                "from payment_info pi " +
                "join dwd_trade_order_detail od " +
                "on pi.order_id = od.order_id " +
                "join `base_dic` for system_time as of pi.pt as dic " +
                "on pi.payment_type = dic.dic_code");

        // 将resultTable注册为视图
        tableEnv.createTemporaryView("result_table_base", resultTable);

        // 将创建视图的代码修改为：
        tableEnv.executeSql("" +
                "create temporary view result_table_with_event_time as " +  // 修改视图名称
                "select " +
                "order_detail_id, " +
                "order_id, " +
                "user_id, " +
                "sku_id, " +
                "sku_name, " +
                "province_id, " +
                "activity_id, " +
                "activity_rule_id, " +
                "coupon_id, " +
                "payment_type_code, " +
                "payment_type_name, " +
                "callback_time, " +
                "source_id, " +
                "source_type_id, " +
                "source_type_name, " +
                "sku_num, " +
                "order_price, " +
                "split_activity_amount, " +
                "split_coupon_amount, " +
                "split_payment_amount, " +
                "row_op_ts, " +
                "TO_TIMESTAMP(callback_time) as event_time " +
                "from result_table_base");

        // 创建详细数据的Kafka输出表（带事件时间属性）
        tableEnv.executeSql("" +
                "create table dwd_trade_pay_detail_suc (" +
                "order_detail_id string, " +
                "order_id string, " +
                "user_id string, " +
                "sku_id string, " +
                "sku_name string, " +
                "province_id string, " +
                "activity_id string, " +
                "activity_rule_id string, " +
                "coupon_id string, " +
                "payment_type_code string, " +
                "payment_type_name string, " +
                "callback_time string, " +
                "source_id string, " +
                "source_type_id string, " +
                "source_type_name string, " +
                "sku_num string, " +
                "order_price string, " +
                "split_activity_amount string, " +
                "split_coupon_amount string, " +
                "split_payment_amount string, " +
                "row_op_ts timestamp_ltz(3), " +
                "event_time AS CAST(TO_TIMESTAMP(callback_time) AS TIMESTAMP_LTZ(3)), " +
                "WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND " +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_pay_detail_suc2", "pay_detail_suc_detail_211126"));

        // 插入详细数据到Kafka
        tableEnv.executeSql("" +
                "insert into dwd_trade_pay_detail_suc " +
                "select " +
                "order_detail_id, order_id, user_id, sku_id, sku_name, province_id, " +
                "activity_id, activity_rule_id, coupon_id, payment_type_code, " +
                "payment_type_name, callback_time, source_id, source_type_id, " +
                "source_type_name, sku_num, order_price, split_activity_amount, " +
                "split_coupon_amount, split_payment_amount, row_op_ts " +
                "from result_table_with_event_time");

        // 创建统计结果的Kafka输出表（新增这部分代码）
        tableEnv.executeSql("" +
                "create table kafka_pay_suc_stats (" +
                "start_time STRING, " +
                "end_time STRING, " +
                "pay_suc_uc BIGINT, " +
                "pay_suc_amount DOUBLE " +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_pay_suc_stats2", "pay_suc_stats_211126"));

        // 窗口查询直接使用表的事件时间属性
        tableEnv.executeSql("" +
                "insert into kafka_pay_suc_stats " +
                "select " +
                "DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') as start_time, " +
                "DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') as end_time, " +
                "COUNT(DISTINCT user_id) as pay_suc_uc, " +
                "SUM(CAST(split_payment_amount AS DOUBLE)) as pay_suc_amount " +
                "from TABLE(TUMBLE(TABLE dwd_trade_pay_detail_suc, DESCRIPTOR(event_time), INTERVAL '1' MINUTE)) " +
                "group by window_start, window_end");

    }
}